package com.sdses.flink.sql;

import com.sdses.flink.udf.PgGeometryTypeConvert;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSQL_PgCDC2PgTest {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.createTemporaryFunction("geometry_type_convert", PgGeometryTypeConvert.class);

        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    code STRING ,\n" +
                "    name STRING,\n" +
                "    id BIGINT,\n" +
                "    shape_length FLOAT,\n" +
                "    shape_area FLOAT,\n" +
                "    the_geom STRING,\n" +
                "    import_time_ TIMESTAMP, \n" +
                "    create_date DATE\n" +
                ") WITH (\n" +
                "  'connector' = 'postgres-cdc',\n" +
                "  'hostname' = '172.21.32.239',\n" +
                "  'port' = '5432',\n" +
                "  'username' = 'postgres',\n" +
                "  'password' = 'Scn2m3Sm#9',\n" +
                "  'database-name' = 'sdses_pg',\n" +
                "  'schema-name' = 'public',\n" +
                "  'table-name' = 'test_gis_erp_community',\n" +
                "  'slot.name' = 'flinksql',\n" +
                "  'decoding.plugin.name' = 'pgoutput'\n" +
                ");");

        tableEnv.executeSql("CREATE TABLE sinkTable (\n" +
                "    code STRING ,\n" +
                "    name STRING,\n" +
                "    id BIGINT,\n" +
                "    shape_length FLOAT,\n" +
                "    shape_area FLOAT,\n" +
                "    the_geom STRING,\n" +
                "    import_time_ TIMESTAMP, \n" +
                "    create_date DATE, \n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:postgresql://172.21.32.239:5432/sdses_pg?stringtype=unspecified',\n" +
                "  'table-name' = 'test_gis_erp_community_sink',\n" +
                "  'username' = 'postgres',\n" +
                "  'password' = 'Scn2m3Sm#9'\n" +
                ");");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

//        tableEnv.executeSql("select code, name, id, shape_length, shape_area, the_geom, import_time_, create_date from sourceTable")
//                        .print();

        tableEnv.executeSql("insert into sinkTable " +
                "select code, name, id, shape_length, shape_area, " +
                "geometry_type_convert(the_geom), " +
                "import_time_, create_date from sourceTable").print();
    }

}
